package com.atguigu.day03;

import com.atguigu.utils.IntSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Example10 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(r -> true)
                .process(new MyProcess())
                .print();



        env.execute();
    }

    public static class MyProcess extends KeyedProcessFunction<Boolean,Integer,StatisticPojo>{
        //声明一个状态变量，作为累加器
        //每个key都会维护自己的状态变量
        private ValueState<StatisticPojo> accumulator;


        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //初始化状态变量
            accumulator = getRuntimeContext().getState(
                    new ValueStateDescriptor<StatisticPojo>("accumulator", Types.POJO(StatisticPojo.class))
            );
        }

        @Override
        public void processElement(Integer value, Context ctx, Collector<StatisticPojo> out) throws Exception {
            if (accumulator.value() == null){
                //第一条数据到达
                accumulator.update(new StatisticPojo(
                        value,value,value,1,(double) value
                ));
            } else {
                //读取状态变量中的值
                StatisticPojo temp = accumulator.value();
                //到达数据 和 累加器中的数据 的累加操作
               Integer min = temp.minValue > value ? value : temp.minValue;
               Integer max = temp.maxValue < value ? value : temp.maxValue;
               Integer sum = temp.sumValue + value;
               Integer count = temp.count+1;
               Double avg = (double) sum / count;
               //更新状态变量
                accumulator.update(new StatisticPojo(
                        min,max,sum,count,avg
                ));

            }
            //将累加器中的值发送出去
            out.collect(accumulator.value());

        }
    }

    //输出统计信息的类
    public static class StatisticPojo{
        public Integer minValue;
        public Integer maxValue;
        public Integer sumValue;
        public Integer count;
        public Double avg;


        public StatisticPojo(Integer minValue, Integer maxValue, Integer sumValue, Integer count, Double avg) {
            this.minValue = minValue;
            this.maxValue = maxValue;
            this.sumValue = sumValue;
            this.count = count;
            this.avg = avg;
        }

        public StatisticPojo() {
        }


        @Override
        public String toString() {
            return "StatisticPojo{" +
                    "minValue=" + minValue +
                    ", maxValue=" + maxValue +
                    ", sumValue=" + sumValue +
                    ", count=" + count +
                    ", avg=" + avg +
                    '}';
        }
    }

}
